package com.zlm.app;

import com.zlm.bean.LoginEvent;
import com.zlm.common.BlacklistWarning;
import com.zlm.common.LoginFailWarning;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;

/**
 * Author: Harbour
 * Date: 2021-05-30 10:59
 * Desc:
 */
public class LoginFailWithCEPAPP {

    public static void main(String[] args) throws Exception {
            // step 1 获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

            // step 2 获取数据源
            URL resource = AdBlacklistApp.class.getResource("/LoginLog.csv");
            DataStream<String> inputDataStream = env.readTextFile(resource.getPath());

            // step 3 转换数据类型
            DataStream<LoginEvent> loginEventDataStream = inputDataStream
                    .map(new MapFunction<String, LoginEvent>() {
                        @Override
                        public LoginEvent map(String value) throws Exception {
                            String[] fields = value.split(",");

                            // Long userId, String ip, String loginState, Long timestamp
                            return new LoginEvent(
                                    Long.parseLong(fields[0]),
                                    fields[1],
                                    fields[2],
                                    Long.parseLong(fields[3])
                            );
                        }
                    })
                    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<LoginEvent>() {
                        @Override
                        public long extractAscendingTimestamp(LoginEvent element) {
                            return element.getTimestamp() * 1000L;
                        }
                    });


            // step 4 编写pattern匹配模式,匹配连续两次登录失败
            Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern.<LoginEvent>begin("FirstFail").where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equalsIgnoreCase(value.getLoginState());
                }
            }).next("SecondFail").where(new SimpleCondition<LoginEvent>() {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equalsIgnoreCase(value.getLoginState());
                }
            }).within(Time.seconds(2));


            // step 5 获取pattern流,选之前模式匹配中需要的流
            CEP.pattern(
                    loginEventDataStream.keyBy(LoginEvent::getUserId),
                    loginFailPattern
            ).select(
                    new LoginFailSelectFunction()
            ).print();

            env.execute();
        }

    private static class LoginFailSelectFunction implements PatternSelectFunction<LoginEvent, LoginFailWarning> {
        @Override
        public LoginFailWarning select(Map<String, List<LoginEvent>> map) throws Exception {
            LoginEvent firstFail = map.get("FirstFail").iterator().next();
            LoginEvent secondFail = map.get("SecondFail").iterator().next();
            return new LoginFailWarning(
                    firstFail.getUserId(),
                    firstFail.getTimestamp(),
                    secondFail.getTimestamp(),
                    "login fail 2 time continuously"
            );
        }
    }
}
